package io.github.junxworks.qt.modules.mm.engine;

import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import javax.sql.DataSource;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;

import io.github.junxworks.ep.sys.service.EpDataService;
import io.github.junxworks.junx.core.exception.BaseRuntimeException;
import io.github.junxworks.junx.core.lang.Initializable;
import io.github.junxworks.junx.core.lifecycle.ThreadService;
import io.github.junxworks.junx.event.EventBus;
import io.github.junxworks.qt.constants.QtConstants;
import io.github.junxworks.qt.modules.mm.dto.MmDatastreamCondition;
import io.github.junxworks.qt.modules.mm.dto.MmVariableCondition;
import io.github.junxworks.qt.modules.mm.service.DatasourceService;
import io.github.junxworks.qt.modules.mm.service.MmDatastreamService;
import io.github.junxworks.qt.modules.mm.service.MmModelService;
import io.github.junxworks.qt.modules.mm.service.MmVariableService;
import io.github.junxworks.qt.modules.mm.vo.MmDatasourceVo;
import io.github.junxworks.qt.modules.mm.vo.MmDatastreamVo;
import io.github.junxworks.qt.modules.mm.vo.MmExpressionVo;
import io.github.junxworks.qt.modules.mm.vo.MmModelVo;
import io.github.junxworks.qt.modules.mm.vo.MmVariableVo;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;

@Service
public class Engine extends ThreadService implements Initializable {
	@Autowired
	private MmModelService modelService;

	@Autowired
	private MmDatastreamService datastreamService;

	@Autowired
	private MmVariableService variableService;

	@Autowired
	private DatasourceService datasourceService;

	@Autowired
	private EpDataService dataService;

	@Autowired
	private DataSource dataSource;

	@Autowired
	private EventBus eventBus;

	private static Long interval = 5 * 60 * 1000l; //5分钟

	private Map<String, ExeContext> exeCtxCache;

	public Map<String, ExeContext> getExeCtxCache() {
		return exeCtxCache;
	}

	public ExeContext execute(ExeContext exec) throws Exception {
		ExeContext ctx = exeCtxCache.get(exec.getUID());
		if (ctx == null) {
			exec.setEventBus(eventBus);
			exeCtxCache.put(exec.getUID(), exec);
			ctx = exec;
		}
		if (ctx.isBusy()) {
			throw new BaseRuntimeException("引擎正在执行中");
		}
		String tradeDate = exec.getTradeDate();
		ctx.work();
		try {
			ctx.setTradeDate(tradeDate); //重置交易日期
			modelService.clearExeResultsByDate(tradeDate); //清空当日所有执行结果，重新统计，之前记录不保留，目前存储太宝贵，只能执行1次删1次
			/***************************************开始执行逻辑******************************************/
			//【*】准备初始化上下文
			if (!ctx.isInitialized()) {
				//初始化模型基础数据，执行初始化根据模型关联出所有数据流
				List<String> models = ctx.getModelNos();
				if (ctx.isOnlyStat()) {
					//如果有模型，还是按模型的统计逻辑走，没有模型按统计逻辑走
					if (models == null || models.isEmpty()) {
						MmVariableCondition condition = new MmVariableCondition();
						condition.setNames(ctx.getStatNames());
						List<MmVariableVo> variables = variableService.queryList(condition);
						ctx.setVariables(variables); //需要执行的统计
						Set<Long> allDs = Sets.newHashSet(variableService.queryAllDatastreamByVarIds(variables.stream().flatMap(v -> {
							return Stream.of(v.getId());
						}).collect(Collectors.toList())));
						if (allDs.isEmpty()) {
							return ctx;
						}
						ctx.addAllDatastream(initDatastreamByIds(allDs));//数据流
					}
				}
				if (models != null && !models.isEmpty()) {
					Set<Long> allDs = modelService.queryAllDatastreams(models);
					if (allDs.isEmpty()) {
						//没有数据流，直接返回
						return ctx;
					}
					//初始化资源
					ctx.addAllDatastream(initDatastreamByIds(allDs));//数据流
					ctx.setVariables(modelService.queryAllVariables(models));//指标
					ctx.setExpressions(modelService.queryExpressions(models));//表达式
					Map<String, List<MmExpressionVo>> expressionMap = Maps.newHashMap();
					models.forEach(m -> {
						expressionMap.put(m, modelService.queryExpressions(Lists.newArrayList(m)));
					});
					ctx.setExpressionMap(expressionMap);
					ctx.setModelMap(modelService.queryModels(models).stream().collect(Collectors.toMap(MmModelVo::getModelNo, v -> v)));//模型
				}
				//初始化需要执行的股票
				if (ctx.getTsCodes() == null) {
					try {
						List<Map<String, Object>> codes = dataService.getDataBySQL("select ts_code from d_stock_profile where list_status='L' order by ts_code asc", Maps.newHashMap());
						ctx.setTsCodes(codes.stream().flatMap(c -> {
							return Stream.of(String.valueOf(c.get("ts_code")));
						}).collect(Collectors.toList()));
					} catch (Exception e) {
						throw new BaseRuntimeException("数据库查询失败", e);
					}
				}
				if (ctx.getIndustry() == null && ctx.getKeywords().contains(QtConstants.KEYWORD_INDUSTRY)) {
					try {
						List<Map<String, Object>> codes = dataService.getDataBySQL("select distinct ifnull(industry,'未知') industry from d_stock_profile where list_status='L' order by ts_code asc", Maps.newHashMap());
						ctx.setIndustry(codes.stream().flatMap(c -> {
							return Stream.of(String.valueOf(c.get("industry")));
						}).collect(Collectors.toList()));
					} catch (Exception e) {
						throw new BaseRuntimeException("数据库查询失败", e);
					}
				}
			}
			ctx.initialize();
			for (Datastream ds : ctx.getDataStreams()) {
				ds.initialize();
			}
			new ModelRunner(ctx).run();
			ctx.getCounter().await();
		} finally {
			ctx.done();
		}
		return ctx;
	}

	private List<Datastream> initDatastreamByIds(Set<Long> allDs) {
		MmDatastreamCondition condition = new MmDatastreamCondition();
		condition.setIds(allDs);
		List<MmDatastreamVo> dss = datastreamService.queryList(condition);
		return dss.stream().flatMap(d -> {
			Datastream ds = new Datastream();
			ds.setSqlTemplate(d.getSqlContent());
			ds.setDsName(d.getDatastreamName());
			MmDatasourceVo datasource = datasourceService.queryDatasourceEntityByDsId(d.getDatasourceId());
			ds.setDatasource(datasource);
			ds.setDataSource(dataSource);
			ds.setFields(datastreamService.queryFieldsByDsId(d.getId()));
			ds.setKeyColumn(d.getKeyColumn());
			return Stream.of(ds);
		}).collect(Collectors.toList());
	}

	@Override
	protected void onStart() throws Throwable {
		exeCtxCache = Maps.newConcurrentMap();
	}

	@Override
	protected void onStop() throws Throwable {
		exeCtxCache.values().forEach(ectx -> {
			try {
				ectx.destroy();
			} catch (Exception e) {
			}
		});
		exeCtxCache.clear();
		exeCtxCache = null;
	}

	@Override
	protected void onRun() throws Throwable {
		Thread.sleep(interval);
		//检查是否有上下文dead
		Iterator<Map.Entry<String, ExeContext>> iterator = exeCtxCache.entrySet().iterator();
		while (iterator.hasNext()) {
			Map.Entry<String, ExeContext> entry = iterator.next();
			if (entry.getValue().isDead()) {
				entry.getValue().destroy();
				iterator.remove();
			}
		}
	}

	@Override
	@PostConstruct
	public void initialize() throws Exception {
		this.start();

	}

	@Override
	@PreDestroy
	public void destroy() throws Exception {
		this.stop();
	}
}
